package dataGen;

import org.apache.commons.io.FileUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2022/2/15
 */
public class DataGenToKafka {
    public static void main(String[] args) throws IOException {
        String gen_sql = FileUtils.readFileToString(new File(args[0]), StandardCharsets.UTF_8);
        String kafka_sql = FileUtils.readFileToString(new File(args[1]), StandardCharsets.UTF_8);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env);

        streamTableEnvironment.executeSql(gen_sql);
        streamTableEnvironment.executeSql(kafka_sql);

        Table table = streamTableEnvironment.sqlQuery("select * from gen_table");
        table.executeInsert("kafka_table");

    }
}
